package kafka.util;

/**
 * @author: li chao
 * @Desc:
 * @create: 2024-07-16 14:18
 **/
import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.NewTopic;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class KafkaTopicCreator {
    public static void createTopic(String bootstrapServers, String topicName, int numPartitions, short replicationFactor) {
        Properties props = new Properties();
        props.put("bootstrap.servers", bootstrapServers);
        AdminClient admin = AdminClient.create(props);

        NewTopic topic = new NewTopic(topicName, numPartitions, replicationFactor);
        try {
            admin.createTopics(Collections.singleton(topic)).all().get();
            System.out.println("Topic created successfully");
        } catch (InterruptedException | ExecutionException e) {
            e.printStackTrace();
        } finally {
            admin.close();
        }
    }

    public static void main(String[] args) {
        String bootstrapServers = "192.168.1.102:9092"; // Kafka 服务器地址和端口
        String topicName = "test-topic"; // 要创建的 topic 名称
        int numPartitions = 1; // topic 分区数
        short replicationFactor = 1; // topic 副本因子

        createTopic(bootstrapServers, topicName, numPartitions, replicationFactor);
    }
}